package com.yifeng.repo.controller.rocketmq.message.consume;

import com.talkyun.utils.revised.Looper;
import com.yifeng.repo.base.api.TimeRange;
import com.yifeng.repo.base.constant.BaseConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * Created by daibing on 2023/8/6.
 */
@Slf4j
public class SubscribePullConsumerLooper extends Looper {
    private final String topic;
    private final Collection<MessageQueue> messageQueues;
    private final DefaultLitePullConsumer pullConsumer;
    private final List<TimeRange> timeRangeList;
    private final Consumer<List<MessageExt>> msgListConsumer;
    private long lastPrintTime = 0L;
    private boolean paused = false;

    public SubscribePullConsumerLooper(String name, int interval, int errorInterval,
                                       String topic, Collection<MessageQueue> messageQueues,
                                       DefaultLitePullConsumer pullConsumer, List<TimeRange> timeRangeList,
                                       Consumer<List<MessageExt>> msgListConsumer) {
        super(name, interval, errorInterval);
        this.topic = topic;
        this.messageQueues = messageQueues;
        this.pullConsumer = pullConsumer;
        this.timeRangeList = timeRangeList;
        this.msgListConsumer = msgListConsumer;
        log.info("SubscribePullConsumerLooper ok: name={}, timeRangeList={}", name, timeRangeList);
    }

    @Override
    protected void loop() throws Throwable {
        // 检查当前时间段
        boolean nowInside = TimeRange.nowInside(timeRangeList);
        if (nowInside) {
            // 1.如果已经暂停，就恢复消费，并且设置状态标记
            if (paused) {
                pullConsumer.resume(messageQueues);
                paused = false;
            }

            // 2.拉取&消费消息
            List<MessageExt> messages = pullConsumer.poll();
            msgListConsumer.accept(messages);
            messages.clear();

            // 3. 间歇性打印消费队列信息
            if (lastPrintTime + BaseConstant.ONE_HOUR_MS < System.currentTimeMillis()) {
                log.info("SubscribePullConsumerLooper ok: name={}, timeRangeList={}, assignment={}",
                        getName(), timeRangeList, buildLog(pullConsumer.assignment()));
                lastPrintTime = System.currentTimeMillis();
            }
        } else {
            // 1.如果已经暂停，继续睡眠1分钟
            if (paused) {
                doSleep(60 * 1000);
                return;
            }
            // 2.如果没有暂停，就暂停，并且设置状态标记，以及睡眠1分钟
            pullConsumer.pause(messageQueues);
            paused = true;
            doSleep(60 * 1000);
        }
    }

    @Override
    protected void loopThrowable(Throwable t) {
        log.warn("consumer {} failed: error=", getName(), t);
    }

    @Override
    protected void clear() {
        pullConsumer.unsubscribe(topic);
        pullConsumer.shutdown();
    }

    public DefaultLitePullConsumer getPullConsumer() {
        return pullConsumer;
    }

    private String buildLog(Collection<MessageQueue> messageQueues) {
        Map<String, List<MessageQueue>> topic2MessageQueues = messageQueues.stream().collect(Collectors.groupingBy(MessageQueue::getTopic));
        StringBuilder sb = new StringBuilder();
        topic2MessageQueues.forEach((topic, messageQueueList) -> {
            String ids = messageQueueList.stream()
                    .map(queue -> String.valueOf(queue.getQueueId()))
                    .reduce((a, b) -> a + "," + b)
                    .orElse("NONE");
            sb.append("topic=").append(topic).append(",");
            sb.append("queues=").append(ids).append(",");
        });
        return sb.toString();
    }
}
